package com.shujia.dwd

import java.awt.geom.Point2D
import java.text.SimpleDateFormat

import com.shujia.grid.Grid
import com.shujia.util.SparkTool
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Row, SparkSession}

import scala.collection.immutable.StringOps
import scala.collection.mutable.ListBuffer

object DwdStayPointApp extends SparkTool{
  /**
    * 在子类是实现run方法，实现自定义的代码逻辑
    *
    * import spark.implicits._
    * import org.apache.spark.sql.functions._
    *
    * @param spark
    */
  override def run(spark: SparkSession): Unit = {
    import spark.implicits._
    //1、读取融合表的数据，取出所需要的字段

    val dataDF: DataFrame = spark.sql(
      s"""
        |select
        |mdn, -- 手机号
        |start_time, -- 业务时间，20180503211049,20180503210349
        |county_id, -- 停留点的区县编码
        |grid_id -- 网格编号
        |from dwd.dwd_res_regn_mergelocation_msk_d  where day_id=$day_id
      """.stripMargin)
    dataDF

    //2、按照手机号进行分组

    val kvRDD: RDD[(String, (String, String, String, String))] = dataDF
      .rdd
      .map{
        case Row(mdn: String, start_time: String, county_id: String, grid_id: String) =>{
          //切分业务时间
          val split: Array[String] = start_time.split(",")
          //开始时间
          val startTime: String = split(1)
          //结束时间
          val endTime: String = split(0)

          //因为要根据手机号进行分组，所以使用mdn作为key

          (mdn,(grid_id,startTime,endTime,endTime))
        }
      }

    //使用groupByKey进行分组

    val groupByKeyRDD: RDD[(String, Iterable[(String, String, String, String)])] = kvRDD.groupByKey()

    /**
      * 按照时间顺序对数据进行聚类
      */
    val juleiRDD: RDD[(String, String, String, String, String)] = groupByKeyRDD.flatMap{
      case (mdn: String, points: Iterable[(String, String, String, String)])=>

        //一个人一天的数据,因为要多次使用所以将迭代器转为List
      val pointList: List[(String, String, String, String)] = points.toList

        //按照开始时间进行升序排序
      val sortPoint: List[(String, String, String, String)] = pointList.sortBy(_._2)

        /**
          * 取出第一条数据
          *
          */

        val lastPoint: (String, String, String, String) = sortPoint.head
        var lastGrid: String = lastPoint._1
        var lastStartTime: String = lastPoint._2
        var lastEndTime: String = lastPoint._3
        var lastCountyId: String = lastPoint._4

        //存放最终的结果
        val resultPoint = new ListBuffer[(String, String, String, String, String)]

        //tail 取出不包括第一天数据的所有数据
      sortPoint.tail.foreach{
        case (grid_id: String, startTime: String, endTime: String, county_id: String) =>
          if (lastGrid.equals(grid_id)) {

            //如果第一条和第二条是同一个网格，
            lastEndTime = endTime
          } else {
            //如果网格不一样了，保存前面的数据
            resultPoint += ((mdn, lastGrid, lastStartTime, lastEndTime, lastCountyId))

            //切换网格
            lastGrid = grid_id
            lastStartTime = startTime
            lastEndTime = endTime
            lastCountyId = county_id
          }

      }

        //处理最后一个组
        resultPoint += ((mdn, lastGrid, lastStartTime, lastEndTime, lastCountyId))

        //返回结果
        resultPoint.toList
    }
    /**
      * mdn string comment '用户手机号码'
      * ,longi string comment '网格中心点经度'
      * ,lati string comment '网格中心点纬度'
      * ,grid_id string comment '停留点所在电信内部网格号'
      * ,county_id string comment '停留点区县'
      * ,duration string comment '机主在停留点停留的时间长度（分钟）,lTime-eTime'
      * ,grid_first_time string comment '网格第一个记录位置点时间（秒级）'
      * ,grid_last_time string comment '网格最后一个记录位置点时间（秒级）'
      */

    //整理数据

    val stayPointRDD: RDD[(String, String, String, String, String, String, String, String)] = juleiRDD.map{
      case (mdn: String, grid: String, startTime: String, endTime: String, countyId: String) =>

        //获取网格中心点的经纬度
        val centerPoint: Point2D.Double = Grid.getCenter(grid.toLong)

        val longi: String = centerPoint.getX.formatted("%.4f")
        val lati: String = centerPoint.getY.formatted("%.4f")

        //计算停留时间
        val format = new SimpleDateFormat("yyyyMMddHHmmss")
        val duration: Double = (format.parse(endTime).getTime - format.parse(startTime).getTime) / 60000.0
        (mdn, longi, lati, grid, countyId, duration.formatted("%.1f"), startTime, endTime)
    }

    //保存数据
    saveDataAndAddPartitionWithDay(
      stayPointRDD.toDF(),
      "/daas/motl/dwd/dwd_staypoint_msk_d",
      "dwd.dwd_staypoint_msk_d"
    )
  }
}
